#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <dirent.h> //目录名获取配合DIR
#include <assert.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
#include <fcntl.h> //文件控制操作
#include <sys/types.h> //获取pid等
#include <sys/wait.h> //等待子进程结束
#include <time.h> //时间获取
#include <sys/ipc.h> //进程通信
#include <sys/msg.h> //消息队列
#include <sys/shm.h> //共享内存
#include <signal.h> //信号处理
#include <pthread.h> //线程
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "cJSON.h"
#include"queue.h"
#include"threadpool.h"



int flag = 0;//结束服务器的话用这个




struct msg {
    char operator[10];
    int number1;
    int number2;
	int result;
};

char* fun(struct msg *k) {
	cJSON* json = NULL;
	json = cJSON_CreateObject();
	cJSON_AddNumberToObject(json, "result", k->result);
	char* dest = cJSON_Print(json);
	printf("dest:%s\n", dest);
	return dest;//记得返回dest！！！
}
void fun2(int sig) {
	flag = 1;
	exit(0);
}

//生产者线程
void* producer(void* arg) {
	struct threadpool* p = (struct threadpool*)arg;
	int listenfd = 0;

	while (1) {
		struct exam data;
		listenfd = accept(p->sock, NULL, NULL);
		if (listenfd == -1) {
			perror("accept");
			continue;
		}

		memset(&data, 0, sizeof(data));
		data.listenfd = listenfd;

		pthread_mutex_lock(&p->mutex);  
		push(p->que, &data);          
		pthread_mutex_unlock(&p->mutex); 

		printf("进入队列的listenfd：%d\n", listenfd);
		sem_post(&p->semr); 
	}
}


//子线程
void* consumer(void* arg)
{
	int ret = 0;
	int count = 0;
	struct threadpool* p = (struct threadpool*)arg;
	while (flag == 0)
	{
		struct queue* q = NULL;
		
		sem_wait(&p->semr);//阻塞
		pthread_mutex_lock(&p->mutex);//互斥锁

		printf("\n消费者线程---%ld号正在处理数据：\n", pthread_self() % 100);//获取线程id

		q = front(p->que);
		printf("消费者从队列里面拿出第一个包的listenfd是： %d\n", q->content.listenfd);
		//--------------------------------------------------------------------------------------老师改的开始
		pop(p->que);
		pthread_mutex_unlock(&p->mutex);//释放锁
		printf("睡眠5秒-------------\n");
		sleep(5);
		sem_post(&p->semr);

		while(1)
		{
		   处理listenfd
		}
}






while(1)
{

		//--------------------------------------------------------------------------------------老师改的结束
		ret = recv(q->content.listenfd, q->content.data, 1024, 0);
		if (ret == -1) {
			perror("recv");
			continue;
		}

		//解析json
		struct msg k;
		int ret = 0;
		cJSON* json = NULL;
		json = cJSON_Parse(q->content.data);
		if (json == NULL) {
			printf("解析error\n");
		}

		cJSON* num1 = NULL;
		num1 = cJSON_GetObjectItem(json, "number1");
		if (num1 == NULL) {
			printf("读取error\n");
		}
		if (num1->type == cJSON_Number) {
			k.number1 = num1->valueint;
		}

		cJSON* num2 = NULL;
		num2 = cJSON_GetObjectItem(json, "number2");
		if (num2 == NULL) {
			printf("读取error\n");
		}
		if (num2->type == cJSON_Number) {
			k.number2 = num2->valueint;
		}
		cJSON* op = NULL;
		op = cJSON_GetObjectItem(json, "operator");
		if (op == NULL) {
			printf("读取error\n");
		}
		if (op->type == cJSON_String) {
			strcpy(k.operator, op->valuestring);
			if (strcmp(k.operator, "add") == 0) {
				k.result = k.number1 + k.number2;
				char* dest = fun(&k);
				ret = send(q->content.listenfd, dest, strlen(dest), 0);
				if (ret == -1) {
					perror("send");
				}
			}
			else if (strcmp(k.operator, "sub") == 0) {
				k.result = k.number1 + k.number2;
				char* dest = fun(&k);
				ret = send(q->content.listenfd, dest, strlen(dest), 0);
				if (ret == -1) {
					perror("send");
				}
			}
			else if (strcmp(k.operator, "mul") == 0) {
				k.result = k.number1 + k.number2;
				char* dest = fun(&k);
				ret = send(q->content.listenfd, dest, strlen(dest), 0);
				if (ret == -1) {
					perror("send");
				}
			}
			else if (strcmp(k.operator, "div") == 0) {
				k.result = k.number1 + k.number2;
				char* dest = fun(&k);
				ret = send(q->content.listenfd, dest, strlen(dest), 0);
				if (ret == -1) {
					perror("send");
				}
			}
		}
		
	
	}
	printf("消费者线程---%ld号退出\n", pthread_self() % 100);
}


int main(int argc, char *argv[])
{
	

	int socketfd = 0;
	int sock = 0;
	int ret = 0;
	int listenfd = 0;
	struct sockaddr_in addr;
	sock = socket(AF_INET, SOCK_STREAM, 0);
	if (sock == -1) {
		perror("socket");
		exit(-1);
	}

	addr.sin_family = AF_INET;
	addr.sin_port = htons(6912);//-----------------------------------------------------------------------------------------
	addr.sin_addr.s_addr = inet_addr("192.168.48.128");

	ret = bind(sock, (struct sockaddr*)&addr, sizeof(struct sockaddr));
	if (ret == -1)
	{
		perror("bind");
		exit(-1);
	}
	listen(sock, 5);









	int threads = 0;
	int queue_length = 0;
	threads = 5;//控制消费者线程数量
	queue_length = 10;//队列长度控制的是一个生产者生产的数量						

	printf("线程池中消费者线程的数量:%d\n", threads);
	printf("队列长度:%d\n", queue_length);

	// 创建线程池---多消费者线程
	struct threadpool* p = create_threadpool(threads, queue_length, consumer);
	p->que = create_queue();//队列的头结点
	// 创建生产者线程
	pthread_t tid1 = 0;
	p->queue_length_max = queue_length;
	p->sock = sock;

	printf("主函数的sock是：%d\n", sock);


	sem_init(&p->semw, 0, queue_length);//初始化信号量			
	sem_init(&p->semr, 0, 0);//初始化信号量		
	pthread_mutex_init(&p->mutex, NULL);//初始化互斥锁
	p->que = create_queue();
	ret = pthread_create(&tid1, NULL, producer, p);//激活生产者线程
	if (ret != 0) {
		perror("pthread_create");
	}

	while (flag == 0)
	{
		sleep(1);
	}
	sem_post(&p->semw);//唤醒生产者线程
	//应该唤醒所有消费者线程退出!!!!
	for (int i = 0; i < threads; i++)
	{
		sem_post(&p->semr);//唤醒消费者线程
	}


	pthread_join(tid1, NULL);//等待生产者线程结束
	destroy_threadpool(p, threads);//销毁线程池
	destroy_queue(p->que);//释放队列
	printf("已释放队列及销毁线程池\n");

	return 0;
}